package com.exsun.gateway.jt809.config;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.AbstractMessageListenerContainer;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class KafkaConsumerConfig {
    private static final Logger log = LoggerFactory.getLogger(KafkaConsumerConfig.class);

    @Value("${handlecenter.kafka.consumer.servers}")
    private String servers;
    @Value("${handlecenter.kafka.consumer.enable.auto.commit}")
    private boolean enableAutoCommit;
    @Value("${handlecenter.kafka.consumer.session.timeout}")
    private String sessionTimeout;
    @Value("${handlecenter.kafka.consumer.auto.commit.interval}")
    private String autoCommitInterval;
    @Value("${handlecenter.kafka.consumer.group.id}")
    private String groupId;
    @Value("${handlecenter.kafka.consumer.auto.offset.reset}")
    private String autoOffsetReset;
    @Value("${handlecenter.kafka.consumer.concurrency}")
    private int concurrency;
    @Value("${handlecenter.kafka.consumer.topic}")
    private String HandleCenterKafkaConsumerTopic;
    @Value("${handlecenter.kafka.max.poll.records}")
    private String maxPollRecords;

    public Map<String, Object> consumerConfigs(String group) {
        Map<String, Object> propsMap = new HashMap<String, Object>();
        propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
        /**
         * 如果设置成true,偏移量由auto.commit.interval.ms控制自动提交的频率。
         * 如果设置成false,不需要定时的提交offset，可以自己控制offset，当消息认为已消费过了，这个时候再去提交它们的偏移量。
         * 这个很有用的，当消费的消息结合了一些处理逻辑，这个消息就不应该认为是已经消费的，直到它完成了整个处理。
         */
        propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
        //提交延迟毫秒数
        propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitInterval);
        //执行超时时间
        propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeout);
        propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        //组ID
        propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, group);

        /**
         * 在consumter端配置文件中(或者是ConsumerConfig类参数)有个"autooffset.reset"(在kafka 0.8版本中为auto.offset.reset),
         * 有2个合法的值"largest"/"smallest",默认为"largest",此配置参数表示当此groupId下的消费者,在ZK中没有offset值时(比如新的groupId,或者是zk数据被清空),
         * consumer应该从哪个offset开始消费.largest表示接受接收最大的offset(即最新消息),smallest表示最小offset,即从topic的开始位置消费所有消息.
         */
        propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
        propsMap.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);

        return propsMap;
    }

    public ConsumerFactory<String, String> getConsumerFactory(String group) {
        return new DefaultKafkaConsumerFactory<String, String>(consumerConfigs(group));
    }

    @Bean(name = "HandleCenterBatchRecordConsumerFactory")
    @Primary
    public KafkaListenerContainerFactory<?> batchRecordConsumerFactory() {
        return getConsumerFactory(true,groupId);
    }

    //额外生成group，当同一个topic需要由多个消费者来消费时，需要调用此方法
    public KafkaListenerContainerFactory<?> batchRecordConsumerFactoryByGroudid(String group) {
        return getConsumerFactory(true,group);
    }

    @Bean(name = "HandleCenterOneByOneRecordConsumerFactory")
    @Primary
    public KafkaListenerContainerFactory<?> oneByOneRecordConsumerFactory() {
        log.info(" KafkaConsumerConfig  groupId: " + groupId);
        return getConsumerFactory(false,groupId);
    }

    /**
     * 获得消费者工厂
     * @param batchListener 是否设置为批量消费
     * @return
     */
    public KafkaListenerContainerFactory<?> getConsumerFactory(boolean batchListener, String groupId) {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<String, String>();
        factory.setConsumerFactory(getConsumerFactory(groupId));
        factory.setConcurrency(concurrency);
        //设置为批量消费，每个批次数量在Kafka配置参数中设置ConsumerConfig.MAX_POLL_RECORDS_CONFIG
        factory.setBatchListener(batchListener);
        //设置提交偏移量的方式
        factory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.MANUAL_IMMEDIATE);
        return factory;
    }
}
